Dans ce guide, je vais vous montrer une approche étape par étape pour structurer une application Web Flask RESTPlus pour les environnements de test, de développement et de production. J'utiliserai un système d'exploitation basé sur Linux (Ubuntu), mais la plupart des étapes peuvent être répliquées sur Windows et Mac.
Avant de continuer avec ce guide, vous devez avoir une compréhension de base du langage de programmation Python et du micro framework Flask.
Comment structurer un service Web Flask-RESTPlus pour les builds de production
Configuration et organisation du projet
Modèles de bases de données et migration
Protection et autorisation d'itinéraire
Extension de l'application et conclusion
Nous utiliserons les fonctionnalités et extensions suivantes dans notre projet.
•.Flask-Bcrypt : Une extension Flask qui fournit des utilitaires de hachage bcrypt pour votre application .
•.Flask-Migrate : une extension qui gère les migrations de base de données SQLAlchemy pour les applications Flask utilisant Alembic. Les opérations de base de données sont disponibles via l'interface de ligne de commande Flask ou via l'extension Flask-Script.
•.Flask-SQLAlchemy : Une extension pour Flask qui ajoute la prise en charge de SQLAlchemy à votre application.
•.PyJWT : une bibliothèque Python qui vous permet d'encoder et de décoder des jetons Web JSON (JWT). JWT est une norme industrielle ouverte ( RFC 7519 ) pour représenter les réclamations en toute sécurité entre deux parties.
•.Flask-Script : une extension qui prend en charge l'écriture de scripts externes dans Flask et d'autres tâches de ligne de commande qui appartiennent à l'extérieur de l'application Web elle-même.
•.Espaces de noms ( Blueprints )
•.Flask-restplus
•.Test de l'unité
Flask-RESTPlus est une extension pour Flask qui ajoute la prise en charge de la création rapide d'API REST. Flask-RESTPlus encourage les meilleures pratiques avec une configuration minimale. Il fournit une collection cohérente de décorateurs et d'outils pour décrire votre API et exposer correctement sa documentation (à l'aide de Swagger).
Créez un nouvel environnement et activez-le en exécutant la commande suivante sur le terminal:
cd /home/yannick/media/dplus/python-dev/
python -m venv venvflask
cd venvflask
git clone https://github.com/cosmic-byte/flask-restplus-boilerplate.git
cd ..
source venvflask/bin/activate
cd venvflask/flask-restplus-boilerplate
J'utiliserai une structure fonctionnelle pour organiser les fichiers du projet par ce qu'ils font. Dans une structure fonctionnelle, les modèles sont regroupés dans un répertoire, les fichiers statiques dans un autre et les vues dans un troisième.
Dans le répertoire du projet, créez un nouveau package appelé app. A l'intérieur app, créez deux packages main et test. La structure de votre répertoire devrait ressembler à celle ci-dessous.
.
├── app
│ ├── __init__.py
│ ├── main
│ │ └── __init__.py
│ └── test
│ └── __init__.py
└── requirements.txt
Nous allons utiliser une structure fonctionnelle pour modulariser notre application.
A l' intérieur du mainpaquet, créer trois autres paquets à savoir: controller, serviceet model. Le modelpackage contiendra tous nos modèles de base de données tandis que le servicepackage contiendra toute la logique métier de notre application et enfin le controllerpackage contiendra tous nos points de terminaison d'application. La structure arborescente devrait maintenant ressembler à ceci:
.
├── app
│ ├── __init__.py
│ ├── main
│ │ ├── controller
│ │ │ └── __init__.py
│ │ ├── __init__.py
│ │ ├── model
│ │ │ └── __init__.py
│ │ └── service
│ │ └── __init__.py
│ └── test
│ └── __init__.py
└── requirements.txt
Permet maintenant d'installer les packages requis. Assurez-vous que l'environnement virtuel que vous avez créé est activé et exécutez les commandes suivantes sur le terminal:
pip install flask-bcrypt
pip install flask-restplus
pip install Flask-Migrate
pip install pyjwt
pip install Flask-Script
pip install flask_testing
Créez ou mettez à jour le requirements.txtfichier en exécutant la commande:
pip freeze > requirements.txt
Le fichier requirements.txt généré doit ressembler à celui ci-dessous:
alembic==0.9.6
aniso8601==1.3.0
bcrypt==3.1.4
cffi==1.14.0
click==6.7
coverage==4.4.2
enum-compat==0.0.2
eventlet==0.21.0
Flask==1.1.2
Flask-Bcrypt==0.7.1
Flask-Cors==3.0.3
Flask-Migrate==2.1.1
flask-restplus==0.10.1
Flask-Script==2.0.6
Flask-SocketIO==2.9.3
Flask-SQLAlchemy==2.3.2
Flask-Testing==0.7.1
gem==0.1.12
greenlet==0.4.15
gunicorn==19.7.1
itsdangerous==0.24
Jinja2==2.11.2
jsonschema==2.6.0
Mako==1.0.7
MarkupSafe==1.0
psycopg2==2.8.5
pycparser==2.18
PyJWT==1.7.1
python-dateutil==2.6.1
python-editor==1.0.3
python-engineio==2.0.2
python-socketio==1.8.4
pytz==2017.3
selenium==3.141.0
six==1.11.0
SQLAlchemy==1.2.0
urllib3==1.25.9
Werkzeug==1.0.1
Dans le mainpackage, créez un fichier appelé config.pyavec le contenu suivant:
import os
# uncomment the line below for postgres database url from environment variable
# postgres_local_base = os.environ['DATABASE_URL']
basedir = os.path.abspath(os.path.dirname(__file__))
class Config:
SECRET_KEY = os.getenv('SECRET_KEY', 'my_precious_secret_key')
DEBUG = False
class DevelopmentConfig(Config):
# uncomment the line below to use postgres
# SQLALCHEMY_DATABASE_URI = postgres_local_base
DEBUG = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'flask_boilerplate_main.db')
SQLALCHEMY_TRACK_MODIFICATIONS = False
class TestingConfig(Config):
DEBUG = True
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'flask_boilerplate_test.db')
PRESERVE_CONTEXT_ON_EXCEPTION = False
SQLALCHEMY_TRACK_MODIFICATIONS = False
class ProductionConfig(Config):
DEBUG = False
# uncomment the line below to use postgres
# SQLALCHEMY_DATABASE_URI = postgres_local_base
config_by_name = dict(
dev=DevelopmentConfig,
test=TestingConfig,
prod=ProductionConfig
)
key = Config.SECRET_KEY
Le fichier de configuration contient trois classes d'installation qui comprend l' environnement testing, developmentet production.
Nous utiliserons le modèle d'usine d'application pour créer notre objet Flask. Ce modèle est très utile pour créer plusieurs instances de notre application avec différents paramètres. Cela facilite la facilité avec laquelle nous basculons entre notre environnement de test, de développement et de production en appelant la create_appfonction avec le paramètre requis.
Dans le __init__.pyfichier à l'intérieur du mainpackage, entrez les lignes de code suivantes:
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_bcrypt import Bcrypt
from .config import config_by_name
db = SQLAlchemy()
flask_bcrypt = Bcrypt()
def create_app(config_name):
app = Flask(__name__)
app.config.from_object(config_by_name[config_name])
db.init_app(app)
flask_bcrypt.init_app(app)
return app
Créons maintenant notre point d'entrée d'application. Dans le répertoire racine du projet, créez un fichier appelé manage.pyavec le contenu suivant:
import os
import unittest
from flask_migrate import Migrate, MigrateCommand
from flask_script import Manager
from app.main import create_app, db
app = create_app(os.getenv('BOILERPLATE_ENV') or 'dev')
app.app_context().push()
manager = Manager(app)
migrate = Migrate(app, db)
manager.add_command('db', MigrateCommand)
@manager.command
def run():
app.run()
@manager.command
def test():
"""Runs the unit tests."""
tests = unittest.TestLoader().discover('app/test', pattern='test*.py')
result = unittest.TextTestRunner(verbosity=2).run(tests)
if result.wasSuccessful():
return 0
return 1
if __name__ == '__main__':
manager.run()
Le code ci-dessus manage.pyfait ce qui suit:
•.line 4et 5importe respectivement les modules migrate et manager (nous utiliserons bientôt la commande migrate).
•.line 9appelle la create_appfonction que nous avons créé d' abord pour créer l'instance d'application avec le paramètre requis à partir de la variable d'environnement qui peut être des éléments suivants - dev, prod, test. Si aucun n'est défini dans la variable d'environnement, la valeur par défaut devest utilisée.
•.line 13et 15instancie le gestionnaire et migre les classes en passant l' appinstance à leurs constructeurs respectifs.
•.Dans line 17, nous transmettons les instances dbet MigrateCommandà l' add_commandinterface de managerpour exposer toutes les commandes de migration de base de données via Flask-Script.
•.line 20et 25marque les deux fonctions comme exécutables à partir de la ligne de commande.
Flask-Migrate expose deux classes, Migrateet MigrateCommand. La Migrateclasse contient toutes les fonctionnalités de l'extension. La MigrateCommandclasse n'est utilisée que si l'on souhaite exposer les commandes de migration de base de données via l'extension Flask-Script.
À ce stade, nous pouvons tester l'application en exécutant la commande ci-dessous dans le répertoire racine du projet.
python manage.py run
Si tout va bien, vous devriez voir quelque chose comme ceci:
Créons maintenant nos modèles. Nous allons utiliser l' dbinstance de sqlalchemy pour créer nos modèles.
L' dbinstance contient toutes les fonctions et les aides à la fois sqlalchemyet sqlalchemy.orm et il fournit une classe appelée Modelqui est une base déclarative qui peut être utilisé pour déclarer les modèles.
Dans le modelpackage, créez un fichier appelé user.pyavec le contenu suivant:
from .. import db, flask_bcrypt
class User(db.Model):
""" User Model for storing user related details """
__tablename__ = "user"
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
email = db.Column(db.String(255), unique=True, nullable=False)
registered_on = db.Column(db.DateTime, nullable=False)
admin = db.Column(db.Boolean, nullable=False, default=False)
public_id = db.Column(db.String(100), unique=True)
username = db.Column(db.String(50), unique=True)
password_hash = db.Column(db.String(100))
@property
def password(self):
raise AttributeError('password: write-only field')
@password.setter
def password(self, password):
self.password_hash = flask_bcrypt.generate_password_hash(password).decode('utf-8')
def check_password(self, password):
return flask_bcrypt.check_password_hash(self.password_hash, password)
def __repr__(self):
return "<User '{}'>".format(self.username)
Le code ci-dessus user.pyfait ce qui suit:
•.line 3:La userclasse hérite de la db.Modelclasse qui déclare la classe comme modèle pour sqlalchemy.
•.line 7through 13crée les colonnes requises pour la table utilisateur.
•.line 21est un setter pour le champ password_hashet il utilise flask-bcryptpour générer un hachage en utilisant le mot de passe fourni.
•.line 24compare un mot de passe donné avec déjà enregistré password_hash.
Now to generate the database table from the user model we just created, we will use migrateCommand through the manager interface. For managerto detect our models, we will have to import theuser model by adding below code to manage.py file:
...
from app.main.model import user
...
Now we can proceed to perform the migration by running the following commands on the project root directory:
1.Initiate a migration folder using init command for alembic to perform the migrations.
python manage.py db init
2. Create a migration script from the detected changes in the model using the migrate command. This doesn’t affect the database yet.
python manage.py db migrate --message 'initial database migration'
3. Apply the migration script to the database by using the upgrade command
python manage.py db upgrade
Si tout fonctionne correctement, vous devriez avoir un nouveau
flask_boilerplate_main.dbfichier de base de données sqlLite généré dans le package principal.
Chaque fois que le modèle de base de données change, répétez les commandes migrateetupgrade
Pour être sûr que la configuration de notre configuration d'environnement fonctionne, écrivons quelques tests pour cela.
Créez un fichier appelé test_config.pydans le package de test avec le contenu ci-dessous:
import os
import unittest
from flask import current_app
from flask_testing import TestCase
from manage import app
from app.main.config import basedir
class TestDevelopmentConfig(TestCase):
def create_app(self):
app.config.from_object('app.main.config.DevelopmentConfig')
return app
def test_app_is_development(self):
self.assertFalse(app.config['SECRET_KEY'] is 'my_precious')
self.assertTrue(app.config['DEBUG'] is True)
self.assertFalse(current_app is None)
self.assertTrue(
app.config['SQLALCHEMY_DATABASE_URI'] == 'sqlite:///' + os.path.join(basedir, 'flask_boilerplate_main.db')
)
class TestTestingConfig(TestCase):
def create_app(self):
app.config.from_object('app.main.config.TestingConfig')
return app
def test_app_is_testing(self):
self.assertFalse(app.config['SECRET_KEY'] is 'my_precious')
self.assertTrue(app.config['DEBUG'])
self.assertTrue(
app.config['SQLALCHEMY_DATABASE_URI'] == 'sqlite:///' + os.path.join(basedir, 'flask_boilerplate_test.db')
)
class TestProductionConfig(TestCase):
def create_app(self):
app.config.from_object('app.main.config.ProductionConfig')
return app
def test_app_is_production(self):
self.assertTrue(app.config['DEBUG'] is False)
if __name__ == '__main__':
unittest.main()
Exécutez le test à l'aide de la commande ci-dessous:
python manage.py test
Vous devriez obtenir la sortie suivante:
Maintenant, travaillons sur les opérations liées à l'utilisateur suivantes:
•.créer un nouvel utilisateur
•.obtenir un utilisateur enregistré avec son public_id
•.obtenir tous les utilisateurs enregistrés.
Classe de service utilisateur: cette classe gère toute la logique relative au modèle utilisateur.
Dans le servicepackage, créez un nouveau fichier user_service.pyavec le contenu suivant:
import uuid
import datetime
from app.main import db
from app.main.model.user import User
def save_new_user(data):
user = User.query.filter_by(email=data['email']).first()
if not user:
new_user = User(
public_id=str(uuid.uuid4()),
email=data['email'],
username=data['username'],
password=data['password'],
registered_on=datetime.datetime.utcnow()
)
save_changes(new_user)
response_object = {
'status': 'success',
'message': 'Successfully registered.'
}
return response_object, 201
else:
response_object = {
'status': 'fail',
'message': 'User already exists. Please Log in.',
}
return response_object, 409
def get_all_users():
return User.query.all()
def get_a_user(public_id):
return User.query.filter_by(public_id=public_id).first()
def save_changes(data):
db.session.add(data)
db.session.commit()
Le code ci-dessus user_service.pyfait ce qui suit:
•.line 8through 29crée un nouvel utilisateur en vérifiant d'abord si l'utilisateur existe déjà; il retourne un succès response_objectsi l'utilisateur n'existe pas sinon il retourne un code d'erreur 409et un échec response_object.
•.line 33et 37retourner une liste de tous les utilisateurs enregistrés et un objet utilisateur en fournissant public_idrespectivement.
•.line 40pour 42valider les modifications apportées à la base de données.
Pas besoin d'utiliser jsonify pour formater un objet en JSON, Flask-restplus le fait automatiquement
In the main package, create a new package called util . This package will contain all the necessary utilities we might need in our application.
In the util package, create a new file dto.py. As the name implies, the data transfer object (DTO) will be responsible for carrying data between processes. In our own case, it will be used for marshaling data for our API calls. We will understand this better as we proceed.
from flask_restplus import Namespace, fields
class UserDto:
api = Namespace('user', description='user related operations')
user = api.model('user', {
'email': fields.String(required=True, description='user email address'),
'username': fields.String(required=True, description='user username'),
'password': fields.String(required=True, description='user password'),
'public_id': fields.String(description='user Identifier')
})
The above code within dto.py does the following:
•.line 5 creates a new namespace for user related operations. Flask-RESTPlus provides a way to use almost the same pattern as Blueprint. The main idea is to split your app into reusable namespaces. A namespace module will contain models and resources declaration.
•.line 6crée un nouvel utilisateur dto via l' modelinterface fournie par l' apiespace de noms dans line 5.
Contrôleur utilisateur: la classe contrôleur utilisateur gère toutes les requêtes HTTP entrantes relatives à l'utilisateur.
Sous le controllerpackage, créez un nouveau fichier appelé user_controller.pyavec le contenu suivant:
from flask import request
from flask_restplus import Resource
from ..util.dto import UserDto
from ..service.user_service import save_new_user, get_all_users, get_a_user
api = UserDto.api
_user = UserDto.user
@api.route('/')
class UserList(Resource):
@api.doc('list_of_registered_users')
@api.marshal_list_with(_user, envelope='data')
def get(self):
"""List all registered users"""
return get_all_users()
@api.response(201, 'User successfully created.')
@api.doc('create a new user')
@api.expect(_user, validate=True)
def post(self):
"""Creates a new User """
data = request.json
return save_new_user(data=data)
@api.route('/<public_id>')
@api.param('public_id', 'The User identifier')
@api.response(404, 'User not found.')
class User(Resource):
@api.doc('get a user')
@api.marshal_with(_user)
def get(self, public_id):
"""get a user given its identifier"""
user = get_a_user(public_id)
if not user:
api.abort(404)
else:
return user
line 1through 8importe toutes les ressources requises pour le contrôleur utilisateur.
Nous avons défini deux classes concrètes dans notre contrôleur utilisateur qui sont
userListet user. Ces deux classes étendent la ressource abstraite flask-restplus.
Les ressources concrètes doivent s'étendre à partir de cette classe et exposer des méthodes pour chaque méthode HTTP prise en charge. Si une ressource est invoquée avec une méthode HTTP non prise en charge, l'API renvoie une réponse avec le statut 405 Méthode non autorisée. Sinon, la méthode appropriée est appelée et transmet tous les arguments de la règle d'URL utilisée lors de l'ajout de la ressource à une instance d'API.
L' apiespace de noms line 7ci-dessus fournit au contrôleur plusieurs décorateurs qui incluent, mais sans s'y limiter:
•.api. route : Un décorateur pour acheminer les ressources
•.api. marshal_with : Un décorateur spécifiant les champs à utiliser pour la sérialisation (C'est là que nous utilisons le que userDtonous avons créé précédemment)
•.api. marshal_list_with : Un raccourci décorateur pour marshal_withci-dessus avecas_list = True
•.api. doc : Un décorateur pour ajouter de la documentation api à l'objet décoré
•.api. réponse: un décorateur pour spécifier l'une des réponses attendues
•.api. attendre: Un décorateur pour spécifier le modèle d'entrée attendu (nous utilisons toujours le userDtopour l'entrée attendue)
•.api. param: Un décorateur pour spécifier l'un des paramètres attendus
Nous avons maintenant défini notre espace de noms avec le contrôleur utilisateur. Il est maintenant temps de l'ajouter au point d'entrée de l'application.
Dans le __init__.pyfichier de apppackage, entrez les informations suivantes:
# app/__init__.py
from flask_restplus import Api
from flask import Blueprint
from .main.controller.user_controller import api as user_ns
blueprint = Blueprint('api', __name__)
api = Api(blueprint,
title='FLASK RESTPLUS API BOILER-PLATE WITH JWT',
version='1.0',
description='a boilerplate for flask restplus web service'
)
api.add_namespace(user_ns, path='/user')
Le code ci-dessus blueprint.pyfait ce qui suit:
•.Dans line 8, nous créons une instance de plan en passant nameet import_name. APIest le principal point d'entrée pour les ressources d'application et doit donc être initialisé avec le blueprintin line 10.
•.Dans line 16, nous ajoutons l'espace user_nsde noms utilisateur à la liste des espaces de noms de l' APIinstance.
Nous avons maintenant défini notre plan. Il est temps de l'enregistrer sur notre application Flask.
Mettez manage.pyà jour en l'important blueprintet en l'enregistrant avec l'instance d'application Flask.
from app import blueprint
...
app = create_app(os.getenv('BOILERPLATE_ENV') or 'dev')
app.register_blueprint(blueprint)
app.app_context().push()
...
Nous pouvons maintenant tester notre application pour voir que tout fonctionne bien.
python manage.py run
Ouvrez maintenant l'URL http://127.0.0.1:5000dans votre navigateur. Vous devriez voir la documentation de swagger.
Testons la création d'un nouveau point de terminaison utilisateur à l'aide de la fonctionnalité de test de swagger.
Vous devriez obtenir la réponse suivante
Créons un modèle blacklistTokenpour stocker des jetons sur liste noire. Dans le modelspackage, créez un blacklist.pyfichier avec le contenu suivant:
from .. import db
import datetime
class BlacklistToken(db.Model):
"""
Token Model for storing JWT tokens
"""
__tablename__ = 'blacklist_tokens'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
token = db.Column(db.String(500), unique=True, nullable=False)
blacklisted_on = db.Column(db.DateTime, nullable=False)
def __init__(self, token):
self.token = token
self.blacklisted_on = datetime.datetime.now()
def __repr__(self):
return '<id: token: {}'.format(self.token)
@staticmethod
def check_blacklist(auth_token):
# check whether auth token has been blacklisted
res = BlacklistToken.query.filter_by(token=str(auth_token)).first()
if res:
return True
else:
return False
N'oublions pas de migrer les modifications pour prendre effet sur notre base de données.
Importez la blacklistclasse dans manage.py.
from app.main.model import blacklist
Exécutez les commandes migrateetupgrade
python manage.py db migrate --message 'add blacklist table'
python manage.py db upgrade
Créez ensuite blacklist_service.pydans le package de services avec le contenu suivant pour mettre un jeton sur liste noire:
from app.main import db
from app.main.model.blacklist import BlacklistToken
def save_token(token):
blacklist_token = BlacklistToken(token=token)
try:
# insert the token
db.session.add(blacklist_token)
db.session.commit()
response_object = {
'status': 'success',
'message': 'Successfully logged out.'
}
return response_object, 200
except Exception as e:
response_object = {
'status': 'fail',
'message': e
}
return response_object, 200
Mettez à jour le usermodèle avec deux méthodes statiques pour l'encodage et le décodage des jetons. Ajoutez les importations suivantes:
import datetime
import jwt
from app.main.model.blacklist import BlacklistToken
from ..config import key
•.Codage
def encode_auth_token(self, user_id):
"""
Generates the Auth Token
:return: string
"""
try:
payload = {
'exp': datetime.datetime.utcnow() + datetime.timedelta(days=1, seconds=5),
'iat': datetime.datetime.utcnow(),
'sub': user_id
}
return jwt.encode(
payload,
key,
algorithm='HS256'
)
except Exception as e:
return e
•.Décodage: le jeton sur liste noire, le jeton expiré et le jeton non valide sont pris en compte lors du décodage du jeton d'authentification.
@staticmethod
def decode_auth_token(auth_token):
"""
Decodes the auth token
:param auth_token:
:return: integer|string
"""
try:
payload = jwt.decode(auth_token, key)
is_blacklisted_token = BlacklistToken.check_blacklist(auth_token)
if is_blacklisted_token:
return 'Token blacklisted. Please log in again.'
else:
return payload['sub']
except jwt.ExpiredSignatureError:
return 'Signature expired. Please log in again.'
except jwt.InvalidTokenError:
return 'Invalid token. Please log in again.'
Écrivons maintenant un test pour le usermodèle pour nous assurer que nos fonctions encodeet decodefonctionnent correctement.
Dans le testpackage, créez un base.pyfichier avec le contenu suivant:
from flask_testing import TestCase
from app.main import db
from manage import app
class BaseTestCase(TestCase):
""" Base Tests """
def create_app(self):
app.config.from_object('app.main.config.TestingConfig')
return app
def setUp(self):
db.create_all()
db.session.commit()
def tearDown(self):
db.session.remove()
db.drop_all()
Le BaseTestCaseconfigure notre environnement de test prêt avant et après chaque scénario de test qui le prolonge.
Créez test_user_medol.pyavec les cas de test suivants:
import unittest
import datetime
from app.main import db
from app.main.model.user import User
from app.test.base import BaseTestCase
class TestUserModel(BaseTestCase):
def test_encode_auth_token(self):
user = User(
email='test@test.com',
password='test',
registered_on=datetime.datetime.utcnow()
)
db.session.add(user)
db.session.commit()
auth_token = user.encode_auth_token(user.id)
self.assertTrue(isinstance(auth_token, bytes))
def test_decode_auth_token(self):
user = User(
email='test@test.com',
password='test',
registered_on=datetime.datetime.utcnow()
)
db.session.add(user)
db.session.commit()
auth_token = user.encode_auth_token(user.id)
self.assertTrue(isinstance(auth_token, bytes))
self.assertTrue(User.decode_auth_token(auth_token.decode("utf-8") ) == 1)
if __name__ == '__main__':
unittest.main()
Exécutez le test avec python manage.py test. Tous les tests devraient réussir.
Créons les points de terminaison d'authentification pour la connexion et la déconnexion .
•.Nous avons d'abord besoin d'un dtopour la charge utile de connexion. Nous utiliserons l'auth dto pour l' @expectannotation en loginendpoint. Ajoutez le code ci-dessous audto.py
class AuthDto:
api = Namespace('auth', description='authentication related operations')
user_auth = api.model('auth_details', {
'email': fields.String(required=True, description='The email address'),
'password': fields.String(required=True, description='The user password '),
})
•.Ensuite, nous créons une classe d'aide à l'authentification pour gérer toutes les opérations liées à l'authentification. Ce auth_helper.pysera dans le package de services et contiendra deux méthodes statiques qui sont login_useretlogout_user
Lorsqu'un utilisateur est déconnecté, le jeton de l'utilisateur est mis sur liste noire, c'est-à-dire que l'utilisateur ne peut pas se reconnecter avec ce même jeton.
from app.main.model.user import User
from ..service.blacklist_service import save_token
class Auth:
@staticmethod
def login_user(data):
try:
# fetch the user data
user = User.query.filter_by(email=data.get('email')).first()
if user and user.check_password(data.get('password')):
auth_token = user.encode_auth_token(user.id)
if auth_token:
response_object = {
'status': 'success',
'message': 'Successfully logged in.',
'Authorization': auth_token.decode()
}
return response_object, 200
else:
response_object = {
'status': 'fail',
'message': 'email or password does not match.'
}
return response_object, 401
except Exception as e:
print(e)
response_object = {
'status': 'fail',
'message': 'Try again'
}
return response_object, 500
@staticmethod
def logout_user(data):
if data:
auth_token = data.split(" ")[1]
else:
auth_token = ''
if auth_token:
resp = User.decode_auth_token(auth_token)
if not isinstance(resp, str):
# mark the token as blacklisted
return save_token(token=auth_token)
else:
response_object = {
'status': 'fail',
'message': resp
}
return response_object, 401
else:
response_object = {
'status': 'fail',
'message': 'Provide a valid auth token.'
}
return response_object, 403
•.Créons maintenant des points de terminaison loginet des logoutopérations.
Dans le package du contrôleur, créez
auth_controller.pyavec le contenu suivant:
from flask import request
from flask_restplus import Resource
from app.main.service.auth_helper import Auth
from ..util.dto import AuthDto
api = AuthDto.api
user_auth = AuthDto.user_auth
@api.route('/login')
class UserLogin(Resource):
"""
User Login Resource
"""
@api.doc('user login')
@api.expect(user_auth, validate=True)
def post(self):
# get the post data
post_data = request.json
return Auth.login_user(data=post_data)
@api.route('/logout')
class LogoutAPI(Resource):
"""
Logout Resource
"""
@api.doc('logout a user')
def post(self):
# get auth token
auth_header = request.headers.get('Authorization')
return Auth.logout_user(data=auth_header)
•.À ce stade, la seule chose qui reste est d'enregistrer l' apiespace de noms d' authentification avec l'applicationBlueprint
Mettre à jour le __init__.pyfichier du apppackage avec les éléments suivants
# app/__init__.py
from flask_restplus import Api
from flask import Blueprint
from .main.controller.user_controller import api as user_ns
from .main.controller.auth_controller import api as auth_ns
blueprint = Blueprint('api', __name__)
api = Api(blueprint,
title='FLASK RESTPLUS API BOILER-PLATE WITH JWT',
version='1.0',
description='a boilerplate for flask restplus web service'
)
api.add_namespace(user_ns, path='/user')
api.add_namespace(auth_ns)
Exécutez l'application avec python manage.py runet ouvrez l'url http://127.0.0.1:5000dans votre navigateur.
La documentation swagger doit maintenant refléter l' authespace de noms nouvellement créé avec les points de terminaison loginet logout.
Avant d'écrire quelques tests pour nous assurer que notre authentification fonctionne comme prévu, modifions notre point de terminaison d'enregistrement pour connecter automatiquement un utilisateur une fois l'enregistrement réussi.
Ajoutez la méthode generate_tokenci-dessous pour user_service.py:
def generate_token(user):
try:
# generate the auth token
auth_token = user.encode_auth_token(user.id)
response_object = {
'status': 'success',
'message': 'Successfully registered.',
'Authorization': auth_token.decode()
}
return response_object, 201
except Exception as e:
response_object = {
'status': 'fail',
'message': 'Some error occurred. Please try again.'
}
return response_object, 401
La generate_tokenméthode génère un jeton d' authentification en codant l'utilisateur. id.Ce jeton est renvoyé en réponse.
Ensuite, remplacez le bloc de retour dans la save_new_userméthode ci-dessous
response_object = {
'status': 'success',
'message': 'Successfully registered.'
}
return response_object, 201
avec
return generate_token(new_user)
Il est maintenant temps de tester les fonctionnalités loginet logout. Créez un nouveau fichier de test test_auth.pydans le package de test avec le contenu suivant:
import unittest
import json
from app.test.base import BaseTestCase
def register_user(self):
return self.client.post(
'/user/',
data=json.dumps(dict(
email='example@gmail.com',
username='username',
password='123456'
)),
content_type='application/json'
)
def login_user(self):
return self.client.post(
'/auth/login',
data=json.dumps(dict(
email='example@gmail.com',
password='123456'
)),
content_type='application/json'
)
class TestAuthBlueprint(BaseTestCase):
def test_registered_user_login(self):
""" Test for login of registered-user login """
with self.client:
# user registration
user_response = register_user(self)
response_data = json.loads(user_response.data.decode())
self.assertTrue(response_data['Authorization'])
self.assertEqual(user_response.status_code, 201)
# registered user login
login_response = login_user(self)
data = json.loads(login_response.data.decode())
self.assertTrue(data['Authorization'])
self.assertEqual(login_response.status_code, 200)
def test_valid_logout(self):
""" Test for logout before token expires """
with self.client:
# user registration
user_response = register_user(self)
response_data = json.loads(user_response.data.decode())
self.assertTrue(response_data['Authorization'])
self.assertEqual(user_response.status_code, 201)
# registered user login
login_response = login_user(self)
data = json.loads(login_response.data.decode())
self.assertTrue(data['Authorization'])
self.assertEqual(login_response.status_code, 200)
# valid token logout
response = self.client.post(
'/auth/logout',
headers=dict(
Authorization='Bearer ' + json.loads(
login_response.data.decode()
)['Authorization']
)
)
data = json.loads(response.data.decode())
self.assertTrue(data['status'] == 'success')
self.assertEqual(response.status_code, 200)
if __name__ == '__main__':
unittest.main()
Jusqu'à présent, nous avons réussi à créer nos points de terminaison, à mettre en œuvre les fonctionnalités de connexion et de déconnexion, mais nos points de terminaison restent non protégés.
Nous avons besoin d'un moyen de définir des règles qui déterminent lequel de nos points de terminaison est ouvert ou nécessite une authentification ou même un privilège d'administrateur.
Nous pouvons y parvenir en créant des décorateurs personnalisés pour nos points de terminaison.
Avant de pouvoir protéger ou autoriser l'un de nos points de terminaison, nous devons connaître l'utilisateur actuellement connecté. Nous pouvons le faire en tirant le Authorization tokende l'en-tête de la demande en cours en utilisant la bibliothèque de flacons. request.Nous décodons ensuite les détails de l'utilisateur du Authorization token.
Dans la Authclasse de auth_helper.pyfichier, ajoutez la méthode statique suivante:
@staticmethod
def get_logged_in_user(new_request):
# get the auth token
auth_token = new_request.headers.get('Authorization')
if auth_token:
resp = User.decode_auth_token(auth_token)
if not isinstance(resp, str):
user = User.query.filter_by(id=resp).first()
response_object = {
'status': 'success',
'data': {
'user_id': user.id,
'email': user.email,
'admin': user.admin,
'registered_on': str(user.registered_on)
}
}
return response_object, 200
response_object = {
'status': 'fail',
'message': resp
}
return response_object, 401
else:
response_object = {
'status': 'fail',
'message': 'Provide a valid auth token.'
}
return response_object, 401
Maintenant que nous pouvons récupérer l'utilisateur connecté à partir de la demande, allons-y et créons le decorators.
Créez un fichier decorator.pydans le utilpackage avec le contenu suivant:
from functools import wraps
from flask import request
from app.main.service.auth_helper import Auth
def token_required(f):
@wraps(f)
def decorated(*args, **kwargs):
data, status = Auth.get_logged_in_user(request)
token = data.get('data')
if not token:
return data, status
return f(*args, **kwargs)
return decorated
def admin_token_required(f):
@wraps(f)
def decorated(*args, **kwargs):
data, status = Auth.get_logged_in_user(request)
token = data.get('data')
if not token:
return data, status
admin = token.get('admin')
if not admin:
response_object = {
'status': 'fail',
'message': 'admin token required'
}
return response_object, 401
return f(*args, **kwargs)
return decorated
Pour plus d'informations sur les décorateurs et comment les créer, consultez ce lien .
Maintenant que nous avons créé les décorateurs token_requiredet admin_token_requiredpour un jeton valide et pour un jeton d'administration respectivement, il ne reste plus qu'à annoter les points de terminaison que nous souhaitons protéger avec le décorateur freecodecamp orgappropriate .
Actuellement, pour effectuer certaines tâches dans notre application, nous devons exécuter différentes commandes pour démarrer l'application, exécuter des tests, installer des dépendances, etc. Nous pouvons automatiser ces processus en organisant toutes les commandes dans un fichier en utilisant Makefile.
Dans le répertoire racine de l'application, créez un Makefilesans extension de fichier. Le fichier doit contenir les éléments suivants:
.PHONY: clean system-packages python-packages install tests run all
clean:
find . -type f -name '*.pyc' -delete
find . -type f -name '*.log' -delete
system-packages:
sudo apt install python-pip -y
python-packages:
pip install -r requirements.txt
install: system-packages python-packages
tests:
python manage.py test
run:
python manage.py run
all: clean install tests run
Voici les options du fichier make.
1.make install : installe les packages système et les packages python
2.make clean : nettoie l'application
3.make tests : exécute tous les tests
4.make run : démarre l'application
5.make all: Permet d' effectuer clean-up, installation, courir tests, et startsl'application.
Il est assez facile de copier la structure d'application actuelle et de l'étendre pour ajouter plus de fonctionnalités / points de terminaison à l'application. Visualisez simplement les itinéraires précédents qui ont été mis en œuvre.
N'hésitez pas à laisser un commentaire si vous avez des questions, des observations ou des recommandations. De plus, si ce message vous a été utile, cliquez sur l'icône de clap pour que d'autres le voient ici et en bénéficient également.
Visitez le référentiel github pour le projet complet.
Merci d'avoir lu et bonne chance!